package com.leonzhangxf.client.register;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @author leonzhangxf
 */
public class ZkServiceLoader implements ServiceLoader {

    private CuratorFramework client;

    public ZkServiceLoader(String connectString) {
        client = CuratorFrameworkFactory.builder()
            .connectString(connectString)
            .retryPolicy(new ExponentialBackoffRetry(1000, 3, 10000))
            .namespace("program_1").build();

        client.start();
    }

    @Override
    public List<Node> loadNode(String serviceId, NodeListChangeCallback callback) throws Exception {
        String serviceIdNode = String.format("/%s", serviceId);
        Stat stat = client.checkExists()
            .creatingParentsIfNeeded()
            .forPath(serviceIdNode);
        if (null == stat) {
            return Collections.emptyList();
        }

        client.getChildren().usingWatcher(new ChildCuratorWatcher(client, serviceIdNode, callback)).forPath(serviceIdNode);

        List<String> children = client.getChildren().forPath(serviceIdNode);
        return toNodes(children);
    }

    static List<Node> toNodes(List<String> children) {
        return Optional.ofNullable(children).orElse(Collections.emptyList())
            .stream().map(raw -> {
                String[] str = raw.split("_");
                return new Node(str[0], Integer.parseInt(str[1]));
            }).collect(Collectors.toList());
    }

    public static class ChildCuratorWatcher implements CuratorWatcher {

        private final CuratorFramework client;

        private final String serviceIdNode;

        private final NodeListChangeCallback callback;

        public ChildCuratorWatcher(CuratorFramework client, String serviceIdNode, NodeListChangeCallback callback) {
            this.client = client;
            this.serviceIdNode = serviceIdNode;
            this.callback = callback;
        }

        @Override
        public void process(WatchedEvent watchedEvent) throws Exception {
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected
                && watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                List<String> children = client.getChildren().forPath(serviceIdNode);
                System.out.println("新的子节点信息" + children);

                if (null != callback) {
                    callback.nodeListChanged(toNodes(children));
                }

                client.getChildren().usingWatcher(this).forPath(serviceIdNode);
            }
        }
    }
}
